package com.xazenith.kfk.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xazenith.kfk.config.Configuration;
import com.xazenith.kfk.exceptions.MessageSerializeError;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.function.Consumer;

/**
 * 消息发送器
 * @author zhouwei
 * @since 1.0
 * @version 1.0
 */
public final class MessageProducer {

    private final Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    private final KafkaProducer<String, String> producer;

    private final String servers;

    /**
     * 对应kafka topic
     */
    private final String topic;

    public MessageProducer(String bsServers, String topic) {
        Properties properties = Configuration.producerConfig(bsServers);
        this.producer = new KafkaProducer<>(properties);
        this.servers = bsServers;
        this.topic = topic;
    }

    /**
     * 发送一条消息，返回Future对象的处理结果
     *
     * @param key key
     * @param data data
     *
     * @return {@link Future} 允许调用 {@link Future#get()}阻塞当前操作
     */
    public Future<RecordMetadata> sendMessageGetFuture(String key, Object data) {
        return sendMessageGetFuture(key, data, null);
    }

    /**
     * 发送一条消息，返回Future对象的处理结果，也可以在consumer中处理返回结果
     *
     * @param key key
     * @param data data
     * @param consumer 返回值的处理器
     *
     * @return {@link Future} 允许调用 {@link Future#get()}阻塞当前操作
     */
    public Future<RecordMetadata> sendMessageGetFuture(String key, Object data, Consumer<RecordMetadata> consumer) {
        try {
            if (logger.isDebugEnabled()) {
                logger.debug("start to send message to {}, topic: {}, key: {}", this.servers, this.topic, key);
            }
            return producer.send(
                    new ProducerRecord<>(
                            this.topic,
                            key,
                            new ObjectMapper().writeValueAsString(data)
                    ),
                    (metadata, exception) -> {
                        if (Objects.isNull(exception)) {
                            if (Objects.nonNull(consumer)) {
                                consumer.accept(metadata);
                            }
                        }else {
                            logger.error(exception.getMessage(), exception);
                        }
                    }
            );
        } catch (JsonProcessingException e) {
            throw new MessageSerializeError("could not serialize message in key: "+ key + " , data: " + data, e);
        }
    }

    /**
     * 发送一条消息，不处理返回值
     *
     * @param key key
     * @param data data
     */
    public void sendMessage(String key, Object data) {
        sendMessage(key, data, null);
    }

    /**
     * 发送一条消息，在回调函数中处理返回值
     *
     * @param key key
     * @param data data
     * @param consumer 返回值处理器
     */
    public void sendMessage(String key, Object data, Consumer<RecordMetadata> consumer) {
       sendMessageGetFuture(key, data, consumer);
    }

}
